package com.mccree.elasticsearch.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.IndexResponse;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem;
import co.elastic.clients.elasticsearch.core.search.Hit;
import com.mccree.elasticsearch.service.ElasticsearchService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

/**
 * @Author: MccRee
 * @Date: 2022/5/24 15:33
 */
@Slf4j
@Service
public class ElasticsearchServiceImpl implements ElasticsearchService {

    @Resource(name = "esClient")
    private ElasticsearchClient esClient;

    @Override
    public <T> void saveOne(String index, T t) {
        String id = UUID.randomUUID().toString();
        saveOne(index,id,t);
    }

    @Override
    public <T> void saveOne(String index, String id, T t) {
        try {
            id = StringUtils.isBlank(id) ? UUID.randomUUID().toString() : id;
            String finalId = id;
            IndexResponse response = esClient.index(c -> c
                    .index(index)
                    .id(finalId)
                    .document(t)
            );
        } catch (IOException e) {
            log.error("ES_SAVE_ONE_ERROR:",e);
        }
    }

    @Override
    public <E> void bulkData(String index, List<E> data) {
        try {
            BulkRequest.Builder bulkBuilder = new BulkRequest.Builder();
            for (E e : data){
                bulkBuilder.operations(c -> c
                        .index(x -> x
                                .index(index)
                                .id(UUID.randomUUID().toString())
                                .document(e)
                        )
                );
            }
            BulkResponse bulkResponse = esClient.bulk(bulkBuilder.build());
            if (bulkResponse.errors()){
                log.error("ES批量插入数据异常");
                for (BulkResponseItem item: bulkResponse.items()) {
                    if (item.error() != null) {
                        log.error(item.error().reason());
                    }
                }
            }
        }catch (Exception e){
            log.error("ES_BULK_ERROR:",e);
        }
    }

    @Override
    public <E> List<E> searchTerm(String index, int from, int size, String key,String value,Class<E> clazz) {
        try {
            //查询
            SearchResponse<?> searchResponse = esClient.search(
                    c -> c.index(index)
                            .query(q -> q
                                    .term(t -> t
                                            .field(key+".keyword")
                                            .value(value)
                                    )
                            )
                            .from(from)
                            .size(size)
                    , clazz
            );
            //处理数据
            return dataProcess(searchResponse);
        }catch (Exception ex){
            log.error("ES_SEARCH_TERM_ERROR:",ex);
            return null;
        }
    }

    @Override
    public <E> List<E> searchMatch(String index, int from, int size, String key, String value, Class<E> clazz) {
        try {
            //查询
            SearchResponse<?> searchResponse = esClient.search(
                    c -> c.index(index)
                            .query(q -> q
                                    .match(m -> m
                                            .field(key)
                                            .query(value)
                                    )
                            )
                            .from(from)
                            .size(size)
                    , clazz
            );
            //处理数据
            return dataProcess(searchResponse);
        }catch (Exception ex){
            log.error("ES_SEARCH_MATCH_ERROR:",ex);
            return null;
        }
    }

    /**
     * 处理数据
     * @param searchResponse
     * @param <E>
     * @return
     */
    private <E> List<E> dataProcess(SearchResponse<?> searchResponse) {
        List<? extends Hit<?>> hits = searchResponse.hits().hits();
        List<E> tList = new ArrayList<>();
        for (Hit<?> hit : hits){
            Object source = hit.source();
            tList.add((E) source);
        }
        return tList;
    }
}
